package com.alibaba.datax.plugin.reader.httpreader;

import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.unstructuredstorage.reader.ColumnEntry;
import com.alibaba.datax.plugin.unstructuredstorage.reader.UnstructuredStorageReaderUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.math.BigInteger;
import java.net.HttpURLConnection;
import java.net.URL;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.text.SimpleDateFormat;
import java.util.*;

public class HttpReader extends Reader {
	public static final String UTF_8 = "UTF-8";
	public static class Job extends Reader.Job {
		private static final Logger LOG = LoggerFactory.getLogger(Job.class);

		private Configuration originConfig = null;

		@Override
		public void init() {
			this.originConfig = this.getPluginJobConf();
			this.validateParameter();
		}

		private void validateParameter() {
			String url = this.originConfig.getString(Key.URL);
			try {
				if (StringUtils.isBlank(url)||(!(url.startsWith("http://")||url.startsWith("https://")))) {
					throw DataXException
							.asDataXException(
									HttpReaderErrorCode.ILLEGAL_VALUE,
									String.format(
											"您配置的url: [%s] 为空或不是http接口地址.",
											url));
				}
			} catch (SecurityException se) {
				throw DataXException.asDataXException(
						HttpReaderErrorCode.SECURITY_NOT_ENOUGH,
						String.format("url错误 : [%s] ", url), se);
			}
		}

		@Override
		public void prepare() {
		}

		@Override
		public void post() {
		}

		@Override
		public void destroy() {
		}

		// warn: 如果源目录为空会报错，拖空目录意图=>空文件显示指定此意图
		@Override
		public List<Configuration> split(int adviceNumber) {
			LOG.debug("split() begin...");
			List<Configuration> readerSplitConfigs = new ArrayList<Configuration>();

			for (int i=0;i<adviceNumber;i++) {
				Configuration splitedConfig = this.originConfig.clone();
				//splitedConfig.set(Constant.SOURCE_FILES, files);
				readerSplitConfigs.add(splitedConfig);
			}
			LOG.debug("split() ok and end...");
			return readerSplitConfigs;
		}

	}

	public static class Task extends Reader.Task {
		private static Logger LOG = LoggerFactory.getLogger(Task.class);
		private static final String NEWLINE_FLAG = System.getProperty("line.separator", "\n");

		private Configuration readerSliceConfig;
		private String url;
		private String clientId;
		private List<ColumnEntry> columns;
		private String clientSecret;
		private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
		private String timestamp;

		@Override
		public void init() {
			this.readerSliceConfig = this.getPluginJobConf();
			this.url = this.readerSliceConfig.getString(Key.URL, null);
			this.clientId = this.readerSliceConfig.getString(Key.CLIENTID, null);
			this.clientSecret = this.readerSliceConfig.getString(Key.CLIENTSECRET, null);
			this.columns = UnstructuredStorageReaderUtil.getListColumnEntry(readerSliceConfig, Key.COLUMN);
		}

		@Override
		public void prepare() {

		}

		@Override
		public void post() {

		}

		@Override
		public void destroy() {

		}

		@Override
		public void startRead(RecordSender recordSender) {
			LOG.debug("start read source files...");
			int totalPages = 1;
			Map<String,Object> params = new HashMap<>();
			int pageNum = 1;
			params.put("pageNum",pageNum);
			params.put("pageSize",10);
			timestamp = sdf.format(new Date());
			String version = "v_"+timestamp;
			params.put("version",version);
			try {
				do {
					params.put("pageNum",pageNum++);
					String rtn = httpPost(JSON.toJSONString(params));
					LOG.info(rtn);
					JSONObject rtnObj = JSON.parseObject(rtn);
					Integer code = rtnObj.getInteger("code");
					if (!Integer.valueOf(0).equals(code)) {
						throw DataXException.asDataXException(HttpReaderErrorCode.RUNTIME_EXCEPTION, rtn);
					}
					JSONObject data = rtnObj.getJSONObject("data");
					totalPages = data.getIntValue("totalPages");
					JSONArray content = data.getJSONArray("content");
					for (int i = 0; i < content.size(); i++) {
						JSONObject row = content.getJSONObject(i);
						String[] rowData = new String[this.columns.size()];
						for (int j = 0; j < this.columns.size(); j++) {
							ColumnEntry column = this.columns.get(j);
							rowData[j] = row.get(column.getValue()).toString();
						}
						UnstructuredStorageReaderUtil.transportOneRecord(recordSender,
								columns, rowData, "null", this.getTaskPluginCollector());
					}
					recordSender.flush();
				}while (pageNum>totalPages);
			} catch (IOException e) {
				throw DataXException.asDataXException(HttpReaderErrorCode.RUNTIME_EXCEPTION, e);
			}
			LOG.debug("end read source files...");
		}

		private String httpPost(String data) throws IOException {
			LOG.info(this.url);
			LOG.info(data);
			URL url = new URL(this.url);
			HttpURLConnection conn = (HttpURLConnection)url.openConnection();
			conn.setDoInput(true);// 允许输入
			conn.setDoOutput(true);// 允许输出
			conn.setUseCaches(false); // 不允许使用缓存
			conn.setRequestMethod("POST");
			conn.setRequestProperty("connection", "keep-alive");
			conn.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
			conn.setRequestProperty(Key.CLIENTID,this.clientId);
			//String timestamp = sdf.format(new Date());
			conn.setRequestProperty(Key.TIMESTAMP,timestamp);
			String Authorization = stringToMD5(clientId+clientSecret+timestamp);
			conn.setRequestProperty(Key.AUTHORIZATION,Authorization);
			conn.getOutputStream().write(data.getBytes(UTF_8));
			conn.getOutputStream().flush();
			BufferedReader br = new BufferedReader(new InputStreamReader(conn.getInputStream()));
			StringBuilder sb = new StringBuilder();
			br.lines().forEach(v1->sb.append(v1).append(NEWLINE_FLAG));
			return sb.toString();
		}

		public static String stringToMD5(String plainText) {
			byte[] secretBytes = null;
			try {
				secretBytes = MessageDigest.getInstance("md5").digest(
						plainText.getBytes(UTF_8));
			} catch (NoSuchAlgorithmException e) {
				throw new RuntimeException("没有这个md5算法！");
			} catch (UnsupportedEncodingException e) {
				e.printStackTrace();
			}
			String md5code = new BigInteger(1, secretBytes).toString(16);
			for (int i = 0; i < 32 - md5code.length(); i++) {
				md5code = "0" + md5code;
			}
			return md5code;
		}
	}
}
